fix(server): start event bus processor for manual wiring - #997
Conversation
|
Based on the summary in #997, I adjusted the way Spring beans are constructed in The change are:
In other words, the following construct @Bean
public RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore, QueueManager queueManager,
PushNotificationConfigStore pushConfigStore,
MainEventBus mainEventBus,
PushNotificationSender pushSender,
@Qualifier("a2aInternal") Executor executor) {
ExecutorService eventConsumerExecutor = Executors.newFixedThreadPool(3);
return new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore,
new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager),
executor,
eventConsumerExecutor);
}is transformed into @Bean
public MainEventBusProcessor mainEventBusProcessor(MainEventBus mainEventBus,TaskStore taskStore,
PushNotificationSender pushSender,
QueueManager queueManager) {
return new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager);
}
@Bean
public RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore, QueueManager queueManager,
PushNotificationConfigStore pushConfigStore,
MainEventBusProcessor mainEventBusProcessor,
@Qualifier("a2aInternal") Executor executor) {
return new DefaultRequestHandler(agentExecutor, taskStore, queueManager,
pushConfigStore, mainEventBusProcessor,
executor, executor);
}Spring Boot recognizes Then, test using the following request: {
"jsonrpc": "2.0",
"id": "6e7bf51a-f364-42d7-a060-343da3cc4f43",
"method": "SendMessage",
"params": {
"message": {
"messageId": "messageId-1",
"role": "ROLE_USER",
"parts": [
{
"text": "Can you check systems?"
}
]
}
}
}Response Response Body: {
"error": null,
"id": "6e7bf51a-f364-42d7-a060-343da3cc4f43",
"jsonrpc": "2.0",
"result": {
"role": "ROLE_AGENT",
"parts": [
{
"text": "Hello World",
"metadata": null
}
],
"messageId": "d062faba-39e6-4ac6-b891-9e0ba1363a69",
"contextId": "c2c16e07-dfad-40f1-958c-41d336d4363e",
"taskId": "d38c2bc5-7491-4920-a290-93582d21a9e0",
"referenceTaskIds": null,
"metadata": null,
"extensions": null
}
}That is, a success result was returned. Hopefully, these new improvements will be helpful to your revisions. Thanks for what you did. |
|
Can you please rebase ? Also an example on how you are using this would be great since it is in SpringBoot runtime |
| @SuppressWarnings("NullAway.Init") | ||
| @PostConstruct | ||
| void start() { | ||
| synchronized void start() { |
There was a problem hiding this comment.
I think you can just make this public synchronized
| */ | ||
| public void ensureStarted() { | ||
| // Method intentionally empty - just forces proxy resolution | ||
| start(); |
| // I am unsure about the correct scope. | ||
| // Also reworked to make a Supplier since otherwise the builder gets polluted with wrong tasks | ||
| this.requestContextBuilder = () -> new SimpleRequestContextBuilder(taskStore, false); | ||
| this.mainEventBusProcessor.ensureStarted(); |
There was a problem hiding this comment.
Just call mainEventBusProcessor.start() if we have to.
But I wonder, since this comes in as a constructor parameter, if you could just not call start() on it before calling the constructor? So this line doesn't seem essential to me.
| } | ||
| eventQueue.enqueueEvent(message); | ||
| Message messageToSend = message; | ||
| if (message.taskId() == null || message.contextId() == null) { |
There was a problem hiding this comment.
Claude thinks:
The null-ID backfill in AgentEmitter (lines 522-528 of the diff) is unrelated to the event bus processor startup fix.
The PR's own test doesn't exercise this path — the test executor calls sendMessage(String), not sendMessage(Message).
This should be a separate PR with its own issue reference.
Ensure MainEventBusProcessor.ensureStarted() starts manually constructed processors, and call it from DefaultRequestHandler.create(...). This prevents manually wired integrations from leaving events stuck on the main event bus. Add a regression test for the manual wiring path that returns a Message response. This fixes a2aproject#995
Start the main event bus processor from the public request handler constructor and verify the direct manual wiring path.\n\nThis fixes a2aproject#995
6a59823 to
4b6eac7
Compare
|
I have updated the PR with my comments and rebased on main |
4 participants
Summary
MainEventBusProcessor.start()for CDI and manual wiring.DefaultRequestHandlerconstruction so manual setups process and persist events.@PostConstructand explicit wiring both callstart().Related Issue
This fixes #995.
Verification
mvn -pl server-common -am -Dtest=DefaultRequestHandlerTest#testBuilderStartsManuallyConstructedMainEventBusProcessor -Dsurefire.failIfNoSpecifiedTests=false test— passed (1 test).mvn -pl server-common -am -DskipTests test-compile— passed.